<?php

namespace app\command;

use app\api\controller\wap\Trade;
use app\common\model\coin\CallApi;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Exception;
use think\facade\Db;
use swoole_server;


/**
 * 获取历史火币数据，并进行保存
 * Class HistoryTrade
 * @package app\command
 */
class TradeServer extends Command
{
    /**
     * 初始化的内存表
     */
    const TABLE_LIST = [
        'AUTO_ID'  => ['get_auto_id',true,true,1,[
            ['id','TYPE_INT',4],
        ]],
        'COUNT'  => ['count',true,false,1000,[
            ['last_time','TYPE_STRING',18],
        ]],

        'CALL_LIMIT'  => ['call_limit',true,false,10,[  //记录请求第三方处理
            ['last_time','TYPE_INT',11],
        ]],

    ];


    /**
     * 进程参数设置
     */
    const PROCESS_CONFIG = [
        'TRADE_NUM' => 20,  // api 请求进程数
        'SERVER'    => [
            'REACTOR_NUM'   => 8,   // 最大线程数
            'WORKER_NUM'    => 4,  // 工作进程数
            'TASK_WORKER_NUM'    => 180,  // 异步处理最大进程数
            'MAX_REQUEST'   => 10000,    // 最大请求数
            'DAEMONIZE' => 0,   //是否守护进程
        ],
        'CLIENT'    => [
            'SEND_INTERVAL' =>  10, //单位微妙
        ],

        'LOG'   => [
            'TYPE_NAME_ERROR'  => 'TRADE_ERROR', // 日志类里对应的类型[错误]
            'TYPE_NAME'  => 'TRADE_LOG', // 日志类里对应的类型
            'TYPE_CUSTOM_LIST'  => [
                'TASK'   => 'task',   // 执行任务
                'ADD'   => 'add',   // 添加
                'CALL'   => 'call',   // 请求
                'IGNORE'  => 'ignore',   //忽视
                'RUN_TIME'  => 'time',   //执行时长
            ],  // 自定义类型
        ],
    ];

    /**
     * ENV 配置参数KEY列表
     */
    const ENV_KEY_LIST = [
        'SERVER_UDP_PORT'   => 'trade_server.server_port',
        'SERVER_UDP_HOST'   => 'trade_server.server_host',
    ];

    static $portIndex = null;

    /**
     * 间隔时间执行，单位秒
     */
    const INTERVAL_TIME_LIST = [
        'CALL_API'  => 15,   // 请求api重复限制时间，单位 s

        'CALL_RECORD_DEL'   => 2,   //监控删除请求记录时间
    ];

    /**
     * 休眠等待时间，单位微妙
     */
    const USLEEP_TIME_LIST = [
        'ERROR_RESTART' => 500000,  //进程失败重启时间
    ];


    /**
     * 重写configure
     * {@inheritdoc}
     */
    protected function configure()
    {
        // 命令的名字（"think" 后面的部分）
        $this->setName('TradeServer')
            // 配置一个参数 使用$input->getArgument('username')获取
            ->addArgument('portIndex')  // 端口索引
            // 运行 "php think list" 时的简短描述
            ->setDescription('监控订单状态处理')
            // 运行命令时使用 "--help" 选项时的完整命令描述
            ->setHelp("一直执行 无参数");
    }


    /**
     *  * 重写execute
     *  * {@inheritdoc}
     *
     * @param Input $input
     * @param Output $output
     */
    public function execute(Input $input, Output $output)
    {

        // 设置端口开始索引【增加支持多开处理】
        static::$portIndex = intval($input->getArgument('portIndex'));

        // 创建内存表
        static::swooleTable();

        // 删除请求记录
        $this->delRecord();

        // 开启交互监听
        $this->createServer();

        //开启等待
        Swoole::wait();
    }

    /**
     * 发送消息
     * @param $message 消息内容，随意值，会直接json处理
     * @param int|null $portIndex
     * @return bool
     */
    static function send($message,int $portIndex = null) {
        $client = new \swoole_client(SWOOLE_SOCK_UDP);
        $client->sendTo('127.0.0.1', TradeServer::serverPort($portIndex), json_encode($message));
        usleep(self::PROCESS_CONFIG['CLIENT']['SEND_INTERVAL']);
        $client->close();
        return true;
    }

    /**
     * 定时删除请求记录
     */
    private function delRecord(){
        $this->startProcessAndErrorRestart(function(){
            $nowTime = time();
            $callLimitTable = self::swooleTable('CALL_LIMIT');
            foreach($callLimitTable as $key=>$callRecord) {
                if($nowTime- $callRecord['last_time'] > self::INTERVAL_TIME_LIST['CALL_API'])  {
                    $callLimitTable->del($key);
                }
            }
            sleep(self::INTERVAL_TIME_LIST['CALL_RECORD_DEL']);
        });
    }



    /**
     * 启动服务器
     */
    private function createServer() {
        register_shutdown_function(function(){
            var_dump(error_get_last());
        });
        $port = self::serverPort();
        $host = env(self::ENV_KEY_LIST['SERVER_UDP_HOST']);
        $host = empty($host)?'0.0.0.0':$host;
        // 创建udp服务器
        $serv = new swoole_server($host,$port,SWOOLE_PROCESS, SWOOLE_SOCK_UDP);
        $serv->set([
            'reactor_num'   => self::PROCESS_CONFIG['SERVER']['REACTOR_NUM'],     // reactor thread num
            'worker_num'    => self::PROCESS_CONFIG['SERVER']['WORKER_NUM'],     // worker process num
            'task_worker_num'    => self::PROCESS_CONFIG['SERVER']['TASK_WORKER_NUM'],     // task worker process num
            'max_request'   => self::PROCESS_CONFIG['SERVER']['MAX_REQUEST'],
            'daemonize'   => self::PROCESS_CONFIG['SERVER']['DAEMONIZE'],   //开启守护进程
            'log_level'   => [SWOOLE_LOG_INFO , SWOOLE_LOG_NOTICE , SWOOLE_LOG_WARNING , SWOOLE_LOG_ERROR],
            'log_file'      => dirname(__FILE__) . '/log/TradeServer.log',
        ]);

        //监听数据接收事件
        $serv->on('Packet', function ($serv, $data, $clientInfo) {
            self::log('TASK',$data);
            $serv->task($data);
        });
        $serv->on('Task',function($serv, $task_id, $from_id, $data){
            try {
                // 添加日志
                self::log('ADD',$data);
                if($data = json_decode($data,true)) {
                    if(Crontab::validateWrite($data)) {
                        return true;
                    }
                    $startTime = microtime(true);
                    $this->execReuqest($data);
                    self::log('RUN_TIME',[
                        'time'  => microtime(true) - $startTime,
                        'data'  => $data
                    ]);

                }
            }catch (\Exception $e) {
                Log::write(serialize([
                    'data'  => $data,
                    'err'   => $e->getMessage()
                ]),Log::LOG_TYPE_LIST[self::PROCESS_CONFIG['LOG']['TYPE_NAME_ERROR']],false);
            }
            return true;
        });

        $serv->on('Finish',function($serv,$task_id, $data){

        });
        //启动服务器
        $serv->start();

    }

    /**
     * 添加日志
     * @param $type 类型
     * @param $message 信息
     */
    static function log($type,$message) {
        Log::write(implode(' ',[self::PROCESS_CONFIG['LOG']['TYPE_CUSTOM_LIST'][$type],serialize($message)]),Log::LOG_TYPE_LIST[self::PROCESS_CONFIG['LOG']['TYPE_NAME']],false);
    }

    /**
     * 执行数据请求
     * @param $data 接收的数据
     */
    private function execReuqest($data) {
      //  echo date('Y-m-d H:i:s',time()),"\n";
        $validate = $data['validate'];
        $value = $data['value'];
        $uniqueKey = implode('_',[
            $data['key'],
            $value['type'],
            $value['cover_times'],
            $value['coin_rule_run_id'],
        ]);
        $callLimitTable = self::swooleTable('CALL_LIMIT');
        if(($callRecord = $callLimitTable->get($uniqueKey)) && time() - $callRecord['last_time'] < self::INTERVAL_TIME_LIST['CALL_API']) {
            // 添加日志
            self::log('IGNORE',$data);
            $this->reply($data['portIndex'],$data['key'],$validate,2);    // 返回已调用，等待结果
            return false;
        }
        // 添加日志
        self::log('CALL',$data);
        $callLimitTable->set($uniqueKey,[
            'last_time' => time()
        ]);
        // 调用第三方 api
        $this->callApi($value,$data['portIndex']);
        Db::close();
        $this->reply($data['portIndex'],$data['key'],$validate,2);    // 返回已调用，等待结果
        return true;
    }


    /**
     * 接口对接请求
     * @param $value
     */
    private function callApi($value,$portIndex = null) {
        $params = [
            'coin_rule_id'  => $value['coin_rule_id'],
            'coin_rule_run_id'  => $value['coin_rule_run_id'],
            'cover_times'   => $value['cover_times'],
            'price_min'   => $value['price_min'],
            'price_max'   => $value['price_max'],
            'price_current'   => $value['price_current'],   //当前价
        ];
        try {
            HistoryTrade::takeTime(HistoryTrade::TAKE_TIME_TYPE_LIST['CALL_RUN'],'',$value['ts'],$value['trade-id'],$params['coin_rule_id'],$params['coin_rule_run_id']);
        //    HistoryTrade::takeTime();
            switch($value['type']) {
                case 1:   // 首单
                    $tradeId = (new Trade())->firstOrder($params);
                    break;
                case 2:   // 补仓
                    $tradeId = (new Trade())->coverOrder($params);
                    break;
                case 3:   // 清仓
                    $tradeId = (new Trade())->clearanceOrder($params);
                    break;
            }
            // 初始化前关闭数据库链接
            HistoryTrade::takeTime(HistoryTrade::TAKE_TIME_TYPE_LIST['CALL_END'],'',$value['ts'],$value['trade-id'],$params['coin_rule_id'],$params['coin_rule_run_id'],intval($tradeId)?$tradeId:0);
        } catch (\Exception $e) {   //操作失败处理
            HistoryTrade::takeTime(HistoryTrade::TAKE_TIME_TYPE_LIST['CALL_END'],'',$value['ts'],$value['trade-id'],$params['coin_rule_id'],$params['coin_rule_run_id']);
            return false;
        }
        return true;
    }


    /**
     * 回复请求状态
     * @param $portIndex 请求的端口索引
     * @param $key  键
     * @param $validate 验证值
     * @param $status   状态
     * @return bool
     * @throws Exception
     */
    private function reply($portIndex,$key,$validate,$status) {
        $params = [
            'apiReply',
            $key,
            $validate,
            $status
        ];
        $result = HistoryTrade::tcpTableQuery(implode(' ',$params),$portIndex);
        return true;
    }

    /**
     * 保存日志处理
     * @param array $takeTimeList
     */
    private function takeTime($portIndex,array $takeTimeList) {
        foreach($takeTimeList as $takeTime) {
            HistoryTrade::tcpTableQuery('takeTime ' . implode(' ',$takeTime),$portIndex);
        }
        return true;
    }





    /**
     * 获取tcp的监控端口
     */
    static function serverPort($portIndex = null)
    {
        if(!$port = env(static::ENV_KEY_LIST['SERVER_UDP_PORT'])) {
            $port = 9400;
        }
        // 获取端口
        $port += intval(is_null($portIndex)?static::$portIndex:$portIndex) * 100;
        return $port;
    }



    /**
     * 获取自增ID
     * @param string $tableName 使用到的表名
     * @return mixed 返回自增ID
     */
    static public function getAutoId(string $tableName = 'auto') {

        return  self::swooleTable('AUTO_ID')->lockFunc(function($table) use($tableName){
            if(!$details = $table->get($tableName)) {
                $details = [
                    'id'    => 0,
                ];
            }
            $details['id']++;
            $table->set($tableName,$details);
            return $details['id'];
        });
    }

    /**
     * 失败重启执行函数
     * @param $function
     */
    private function startProcessAndErrorRestart($function,$isMaster = false) {
        Swoole::startProcess(function() use($function){
            while (true) {
                try {
                    // 执行数据处理
                    call_user_func_array($function,[]);
                } catch (\Exception $e) {
                    //       echo $e->getMessage(),':',$e->getLine(),"\n";
                    usleep(self::USLEEP_TIME_LIST['ERROR_RESTART']);
                }
            }
        },true,$isMaster);
    }

    /**
     * 创建及获取表数据
     * @param null $tableKey
     * @param string|null $tableSuffix
     * @return bool|mixed
     */
    private static function swooleTable($tableKey = null,string $tableSuffix = null) {
        if(is_null($tableKey)) {    // 初始化表
            foreach(static::TABLE_LIST as $table) {
                if(!$table[1]) {
                    continue;
                }
                // 创建Swoole表
                Swoole::table($table[0],$table[4],1024 * $table[3],$table[2]);
            }
            return true;
        }

        // 创建Swoole表
        $table = static::TABLE_LIST[$tableKey];
        if($tableSuffix) {
            $table[0] .= $tableSuffix;
        }
        return Swoole::table($table[0],$table[4],1024 * $table[3],$table[2]);
    }

}